{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "958a52c6-da3d-45b5-9c2e-27e579c6066d",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "25/01/17 22:03:34 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "25/01/17 22:03:35 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# Create a SparkSession\n",
    "spark = SparkSession.builder \\\n",
    "    .appName(\"DataFrameSQL\") \\\n",
    "    .master(\"spark://spark-master:7077\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "d3a86cbb-583f-427d-970b-de58f9e7bcf4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "name,age,gender,salary\n",
      "John Doe,30,Male,50000\n",
      "Jane Smith,25,Female,45000\n",
      "David Johnson,35,Male,60000\n",
      "Emily Davis,28,Female,52000\n",
      "Michael Wilson,40,Male,75000\n",
      "Sarah Brown,32,Female,58000\n",
      "Robert Lee,29,Male,51000\n",
      "Lisa Garcia,27,Female,49000\n",
      "James Martinez,38,Male,70000\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "head -10 ../input_files/persons.csv"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "24c56d31-71d2-48c0-8d22-b7eb9c4c347f",
   "metadata": {},
   "source": [
    "### Load Data into a DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "dcd9c1d7-ccc2-4cee-8547-c23f3bee42bb",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/17 22:03:55 WARN TaskSchedulerImpl: Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient resources\n",
      "25/01/17 22:04:10 WARN TaskSchedulerImpl: Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient resources\n",
      "25/01/17 22:04:25 WARN TaskSchedulerImpl: Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient resources\n",
      "25/01/17 22:04:40 WARN TaskSchedulerImpl: Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient resources\n",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Load the synthetic data into a DataFrame\n",
    "data_file_path = \"hdfs://namenode:9000/input_files/persons.csv\"\n",
    "df = spark.read.csv(data_file_path, header=True, inferSchema=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "99b0d2db-5d0e-4e82-9d45-95029f614319",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- name: string (nullable = true)\n",
      " |-- age: integer (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- salary: integer (nullable = true)\n",
      "\n",
      "Initial DataFrame:\n",
      "+------------------+---+------+------+\n",
      "|              name|age|gender|salary|\n",
      "+------------------+---+------+------+\n",
      "|          John Doe| 30|  Male| 50000|\n",
      "|        Jane Smith| 25|Female| 45000|\n",
      "|     David Johnson| 35|  Male| 60000|\n",
      "|       Emily Davis| 28|Female| 52000|\n",
      "|    Michael Wilson| 40|  Male| 75000|\n",
      "|       Sarah Brown| 32|Female| 58000|\n",
      "|        Robert Lee| 29|  Male| 51000|\n",
      "|       Lisa Garcia| 27|Female| 49000|\n",
      "|    James Martinez| 38|  Male| 70000|\n",
      "|Jennifer Rodriguez| 26|Female| 47000|\n",
      "+------------------+---+------+------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Display schema of DataFrame\n",
    "df.printSchema()\n",
    "\n",
    "# Show the initial DataFrame\n",
    "print(\"Initial DataFrame:\")\n",
    "df.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1293db0e-3af1-40d3-9fed-15cb5c6d54f1",
   "metadata": {},
   "source": [
    "### Register the DataFrame as a Temporary Table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "439040dd-8c13-48c9-9a72-acd313da407a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Register the DataFrame as a Temporary Table\n",
    "df.createOrReplaceTempView(\"my_table\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3c0ed5db-3985-4251-935a-3c01edd47005",
   "metadata": {},
   "source": [
    "### Perform SQL-like Queries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "ccf8aa7d-e77f-4fca-93d9-fd4f2581803d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+---+------+------+\n",
      "|              name|age|gender|salary|\n",
      "+------------------+---+------+------+\n",
      "|          John Doe| 30|  Male| 50000|\n",
      "|     David Johnson| 35|  Male| 60000|\n",
      "|       Emily Davis| 28|Female| 52000|\n",
      "|    Michael Wilson| 40|  Male| 75000|\n",
      "|       Sarah Brown| 32|Female| 58000|\n",
      "|        Robert Lee| 29|  Male| 51000|\n",
      "|       Lisa Garcia| 27|Female| 49000|\n",
      "|    James Martinez| 38|  Male| 70000|\n",
      "|Jennifer Rodriguez| 26|Female| 47000|\n",
      "|  William Anderson| 33|  Male| 62000|\n",
      "|   Karen Hernandez| 31|Female| 55000|\n",
      "|Christopher Taylor| 37|  Male| 69000|\n",
      "|     Matthew Davis| 36|  Male| 67000|\n",
      "|    Patricia White| 29|Female| 50000|\n",
      "|     Daniel Miller| 34|  Male| 64000|\n",
      "| Elizabeth Jackson| 30|Female| 52000|\n",
      "|     Joseph Harris| 28|  Male| 53000|\n",
      "|      Linda Martin| 39|Female| 71000|\n",
      "+------------------+---+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Select all rows where age is greater than 25\n",
    "result = spark.sql(\"SELECT * FROM my_table WHERE age > 25\")\n",
    "\n",
    "result.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "8b88f371-fd06-4aed-9fca-bdb4ce1362ff",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----------+\n",
      "|gender|avg_salary|\n",
      "+------+----------+\n",
      "|Female|   52300.0|\n",
      "|  Male|   62100.0|\n",
      "+------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Compute the average salary by gender\n",
    "avg_salary_by_gender = spark.sql(\"SELECT gender, AVG(salary) as avg_salary FROM my_table GROUP BY gender\")\n",
    "avg_salary_by_gender.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "62b21e82-65e2-4b09-b344-a5cc72d9d430",
   "metadata": {},
   "source": [
    "### Creating and managing temporary views."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "6469c424-87cd-4dfa-b015-1c3f8e591de0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a temporary view\n",
    "df.createOrReplaceTempView(\"people\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "9198c861-e79a-4fac-80cc-747c94f062ce",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+---+------+------+\n",
      "|              name|age|gender|salary|\n",
      "+------------------+---+------+------+\n",
      "|          John Doe| 30|  Male| 50000|\n",
      "|     David Johnson| 35|  Male| 60000|\n",
      "|       Emily Davis| 28|Female| 52000|\n",
      "|    Michael Wilson| 40|  Male| 75000|\n",
      "|       Sarah Brown| 32|Female| 58000|\n",
      "|        Robert Lee| 29|  Male| 51000|\n",
      "|       Lisa Garcia| 27|Female| 49000|\n",
      "|    James Martinez| 38|  Male| 70000|\n",
      "|Jennifer Rodriguez| 26|Female| 47000|\n",
      "|  William Anderson| 33|  Male| 62000|\n",
      "|   Karen Hernandez| 31|Female| 55000|\n",
      "|Christopher Taylor| 37|  Male| 69000|\n",
      "|     Matthew Davis| 36|  Male| 67000|\n",
      "|    Patricia White| 29|Female| 50000|\n",
      "|     Daniel Miller| 34|  Male| 64000|\n",
      "| Elizabeth Jackson| 30|Female| 52000|\n",
      "|     Joseph Harris| 28|  Male| 53000|\n",
      "|      Linda Martin| 39|Female| 71000|\n",
      "+------------------+---+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Query the temporary view\n",
    "result = spark.sql(\"SELECT * FROM people WHERE age > 25\")\n",
    "\n",
    "result.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "6d363610-7dc6-4a80-bc2d-c1269691e437",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Drop a temporary view\n",
    "spark.catalog.dropTempView(\"people\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8bcdf631-ee1d-496f-858b-a13d0b43cb18",
   "metadata": {},
   "source": [
    "### Subquries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "5cc8db71-4862-4169-b2d0-7e38ccdf8331",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+\n",
      "| id|   name|\n",
      "+---+-------+\n",
      "|  1|   John|\n",
      "|  2|  Alice|\n",
      "|  3|    Bob|\n",
      "|  4|  Emily|\n",
      "|  5|  David|\n",
      "|  6|  Sarah|\n",
      "|  7|Michael|\n",
      "|  8|   Lisa|\n",
      "|  9|William|\n",
      "+---+-------+\n",
      "\n",
      "+----------+---+------+\n",
      "|department| id|salary|\n",
      "+----------+---+------+\n",
      "|        HR|  1| 60000|\n",
      "|        HR|  2| 55000|\n",
      "|        HR|  3| 58000|\n",
      "|        IT|  4| 70000|\n",
      "|        IT|  5| 72000|\n",
      "|        IT|  6| 68000|\n",
      "|     Sales|  7| 75000|\n",
      "|     Sales|  8| 78000|\n",
      "|     Sales|  9| 77000|\n",
      "+----------+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Create DataFrames\n",
    "employee_data = [\n",
    "    (1, \"John\"), (2, \"Alice\"), (3, \"Bob\"), (4, \"Emily\"),\n",
    "    (5, \"David\"), (6, \"Sarah\"), (7, \"Michael\"), (8, \"Lisa\"),\n",
    "    (9, \"William\")\n",
    "]\n",
    "employees = spark.createDataFrame(employee_data, [\"id\", \"name\"])\n",
    "\n",
    "salary_data = [\n",
    "    (\"HR\", 1, 60000), (\"HR\", 2, 55000), (\"HR\", 3, 58000),\n",
    "    (\"IT\", 4, 70000), (\"IT\", 5, 72000), (\"IT\", 6, 68000),\n",
    "    (\"Sales\", 7, 75000), (\"Sales\", 8, 78000), (\"Sales\", 9, 77000)\n",
    "]\n",
    "salaries = spark.createDataFrame(salary_data, [\"department\", \"id\", \"salary\"])\n",
    "\n",
    "employees.show()\n",
    "\n",
    "salaries.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "871a978f-82f9-4388-afae-77d4b0e7297b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Register as temporary views\n",
    "employees.createOrReplaceTempView(\"employees\")\n",
    "salaries.createOrReplaceTempView(\"salaries\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "c0a3c981-b8d4-409c-bd3a-aaff55ee3661",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+\n",
      "|   name|\n",
      "+-------+\n",
      "|  Emily|\n",
      "|  David|\n",
      "|Michael|\n",
      "|   Lisa|\n",
      "|William|\n",
      "+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Subquery to find employees with salaries above average\n",
    "result = spark.sql(\"\"\"\n",
    "    SELECT name\n",
    "    FROM employees\n",
    "    WHERE id IN (\n",
    "        SELECT id\n",
    "        FROM salaries\n",
    "        WHERE salary > (SELECT AVG(salary) FROM salaries)\n",
    "    )\n",
    "\"\"\")\n",
    "\n",
    "result.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5237a76b-d747-4096-a013-41cc913cd9c0",
   "metadata": {},
   "source": [
    "### Window Function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "afd58076-cada-4552-a676-7919cb17ba2b",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql import functions as F"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "5669387e-aa09-4830-a4c4-c32056e38d5f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---+------+-------+\n",
      "|department| id|salary|   name|\n",
      "+----------+---+------+-------+\n",
      "|        HR|  1| 60000|   John|\n",
      "|        HR|  2| 55000|  Alice|\n",
      "|        HR|  3| 58000|    Bob|\n",
      "|        IT|  4| 70000|  Emily|\n",
      "|        IT|  5| 72000|  David|\n",
      "|        IT|  6| 68000|  Sarah|\n",
      "|     Sales|  7| 75000|Michael|\n",
      "|     Sales|  8| 78000|   Lisa|\n",
      "|     Sales|  9| 77000|William|\n",
      "+----------+---+------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "employee_salary = spark.sql(\"\"\"\n",
    "    select  salaries.*, employees.name\n",
    "    from salaries \n",
    "    left join employees on salaries.id = employees.id\n",
    "\"\"\")\n",
    "\n",
    "employee_salary.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "0f2fb287-69a4-4860-a13b-f76646e2f465",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a window specification\n",
    "window_spec = Window.partitionBy(\"department\").orderBy(F.desc(\"salary\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "dcf9ca4b-f1ce-43f1-ac72-bac31c32e34f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---+------+-------+----+\n",
      "|department| id|salary|   name|rank|\n",
      "+----------+---+------+-------+----+\n",
      "|        HR|  1| 60000|   John|   1|\n",
      "|        HR|  3| 58000|    Bob|   2|\n",
      "|        HR|  2| 55000|  Alice|   3|\n",
      "|        IT|  5| 72000|  David|   1|\n",
      "|        IT|  4| 70000|  Emily|   2|\n",
      "|        IT|  6| 68000|  Sarah|   3|\n",
      "|     Sales|  8| 78000|   Lisa|   1|\n",
      "|     Sales|  9| 77000|William|   2|\n",
      "|     Sales|  7| 75000|Michael|   3|\n",
      "+----------+---+------+-------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Calculate the rank of employees within each department based on salary\n",
    "employee_salary.withColumn(\"rank\", F.rank().over(window_spec)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "7d85a9f5-3da2-4ef7-b201-7bb43ccc6bc2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stop the SparkSession\n",
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fea9d104-e4b3-4be0-9fae-43fdbc391691",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "pyspark",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
